#include "UI_multiplexer.h"

#include "Network.h"

#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/foreach.hpp>
#include <ariba/utility/system/SystemQueue.h>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include "libs/substreambuf/prefix_streambuf.hpp"
#include <boost/array.hpp>

#define foreach BOOST_FOREACH

using namespace std;
using boost::asio::ip::tcp;
using boost::asio::local::stream_protocol; //XXX linux-only
using boost::property_tree::ptree;
using boost::asio::const_buffer;

#define XML_COMMAND_SEPARATOR_LENGTH 2
const static char* XML_COMMAND_SEPARATOR = "<>";

template <class T>
class Stream_connection :
    public boost::enable_shared_from_this< Stream_connection<T> >,
    public Useroutputinterface
{
    typedef Stream_connection<T> self;
public:
    friend class Stream_server<T>;
    
    Stream_connection(boost::asio::io_service& io_service)  :
        sock(io_service),
        valid(true)
    {
    }
    
    virtual ~Stream_connection()
    {
    }
    
    // starts an async_read
    void listen()
    {
        boost::asio::async_read_until(
                this->sock,
                this->sbuf,
                XML_COMMAND_SEPARATOR,
                boost::bind(
                        &self::async_read_handler,
                        this->shared_from_this(),
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred
                )
        );    
    }
    
    
    
    virtual bool write_to_user_virtual(StringPtr s)
    {
        if (!this->valid) {
            return false;
        }
        
        // prepare asio buffer sequence with data and delemiter
        boost::array<const_buffer, 2> data_and_delim = {
                const_buffer(s->data(), s->size()),
                const_buffer(XML_COMMAND_SEPARATOR, XML_COMMAND_SEPARATOR_LENGTH) };
        
        // * async write *
        boost::asio::async_write(
                this->sock,
                data_and_delim,
                boost::bind(
                        &self::async_write_handler,
                        this->shared_from_this(),
                        s,  // makes sure our shared string lives long enough ;-)
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred)
        );
        return true;
    }
    
    
private:
    void async_read_handler(
            const boost::system::error_code& error,
            size_t bytes_transferred)
    {
        cout << "UI_multiplexer: Incoming " << bytes_transferred << " bytes." << endl;//XXX
        
        if (!error)
        {
            // Only parse the substring
            prefix_streambuf sub_buf(
                    &this->sbuf,
                    bytes_transferred - XML_COMMAND_SEPARATOR_LENGTH);
            
            istream stream(&sub_buf);
            ptree* command_pt = new ptree();
            try {
                boost::property_tree::read_xml(stream, *command_pt);
            } catch (boost::property_tree::xml_parser_error& e) {
                StringPtr s(new string("Error: Invalid XML"));
                this->write_to_user(s);
            }
            
            // Remove separator
            this->sbuf.consume(XML_COMMAND_SEPARATOR_LENGTH);
            
            // enqueue in ariba system queue
            ariba::utility::SystemEventType command_event("Command from UI");
            ariba::utility::SystemQueue::instance().scheduleEvent(
                    ariba::utility::SystemEvent(
                            &Network::instance(),
                            command_event,
                            command_pt) );
            
            if (this->valid) {
                // read again
                listen();
            }
        }
    }
    
    void async_write_handler(
            StringPtr s,
            const boost::system::error_code& error,
            size_t bytes_transferred)
    {
        if (error) {
            this->valid = false;
            
            cout << "WARN: async write error." << endl;  // XXX debug output
        }
    }
    
private:
    boost::asio::basic_stream_socket<T> sock;
    boost::asio::streambuf sbuf;
    bool valid;
};



template <class T>
class Stream_server:
    public boost::enable_shared_from_this< Stream_server<T> >
{
    typedef Stream_server<T> self;
public:
    Stream_server(
            boost::asio::io_service& io_service,
            typename boost::asio::basic_socket_acceptor<T>::endpoint_type& endpoint
    )  :
        io_service(io_service),
        acceptor(io_service, endpoint) //TODO exception
    {
    }
    
    
    void accept()
    {
        boost::shared_ptr< Stream_connection<T> > conn(new Stream_connection<T>(io_service));
        
        acceptor.async_accept(
                conn->sock,
                boost::bind(&self::async_accept_handler,
                        this->shared_from_this(),
                        conn,
                        boost::asio::placeholders::error)
        );
    }
    
    
    
private:
    void async_accept_handler(
            boost::shared_ptr< Stream_connection<T> > conn,
            const boost::system::error_code& error
            )
    {
        if ( ! error )
        {
            UI_multiplexer::instance().uis.push_back(conn);
            
            // read
            conn->listen();
        }
        
        // accept further connections
        accept();
    }
    
private:
    boost::asio::io_service& io_service;
    boost::asio::basic_socket_acceptor<T> acceptor;
};

UI_multiplexer::UI_multiplexer()
{
}

UI_multiplexer::~UI_multiplexer()
{
}

bool UI_multiplexer::write_to_user_virtual(StringPtr s)
{
    cout << "TO UIs: " << *s << endl;
    
    bool delivered = false;
    int errors = 0;
    
    for ( vector<UseroutputinterfacePtr>::iterator it = this->uis.begin(); it != this->uis.end(); )
    {
        // * write *
        bool okay = (*it)->write_to_user(s);
        delivered |= okay;   // TODO hauptsache einer okay oder sollen alle okay sein müssen?
        
        // lazy remove invalid uis
        if ( ! okay )
            it = this->uis.erase(it);
        else
            it++;
    }
    
    return delivered;
}

void UI_multiplexer::listen(boost::asio::ip::tcp::endpoint& endpoint)
{
    TcpServerPtr server(new Stream_server<tcp>(this->io_service, endpoint));
    this->tcpServers.push_back(server);
    server->accept();
}

void UI_multiplexer::listen(
        boost::asio::local::stream_protocol::endpoint& endpoint)
{
    UnixServerPtr server(
            new Stream_server<stream_protocol>(this->io_service, endpoint) );
    this->unixServers.push_back(server);
    server->accept();
}

void UI_multiplexer::run()
{
    this->io_service.run();
}

void UI_multiplexer::stop()
{
    // say good bye to all remaining UIs
    StringPtr goodby(new string("Omnibus shutting down."));
    this->write_to_user_virtual(goodby);
    
    usleep(200); // TODO Hier müssten eigentlich synchrone Writes benutzt werden..
    
    // stop all io
    this->io_service.stop();
}
